package com.zy.rpc.consumer.transport.impl;

import com.zy.rpc.common.ask.RpcRequest;
import com.zy.rpc.common.ask.RpcResponse;
import com.zy.rpc.common.codec.RpcDecoder;
import com.zy.rpc.common.codec.RpcEncoder;
import com.zy.rpc.consumer.handle.RpcClientHandle;
import com.zy.rpc.consumer.transport.NettyClientTransportFactory;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;

/**
 * @author zy
 * @date 2023/5/7 16:03
 */
public class NettyClientTransport implements NettyClientTransportFactory {

    private static volatile NettyClientTransport INSTANCE=null;


    private final EventLoopGroup eventLoopGroup;

    private final Bootstrap bootstrap;
//    private final RpcQuestCodec rpcQuestCodec =new RpcQuestCodec();
//    private final ResponseCodec responseCodec =new ResponseCodec();


    public NettyClientTransport() throws InterruptedException {
        System.out.println("客户端启动");
        eventLoopGroup =new NioEventLoopGroup(4);
        bootstrap=new Bootstrap();
        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel channel) throws Exception {
                        channel.pipeline().addLast(new LoggingHandler(LogLevel.TRACE));
                        channel.pipeline().addLast(new RpcEncoder());
                        channel.pipeline().addLast(new RpcDecoder());
                    }
                });
    }

    public static NettyClientTransport getInstance() throws InterruptedException {
        if(INSTANCE==null){
            synchronized (NettyClientTransport.class){
                if (INSTANCE==null){
                    INSTANCE=new NettyClientTransport();
                }
            }
        }
        return INSTANCE;
    }


    @Override
    public  CompletableFuture<RpcResponse> sendRecquest(RpcRequest rpcRequest,String address,Integer port) throws InterruptedException, ExecutionException, TimeoutException {
        CompletableFuture<RpcResponse> future = new CompletableFuture<>();
        ChannelFuture channelFuture = bootstrap.connect(address, port).sync();
        Channel channel = channelFuture.channel();
        channel.pipeline().addLast(new RpcClientHandle(future));
        channel.writeAndFlush(rpcRequest);
        return future;
    }

}
